//package com.qingguo.Test
//
//import java.security.MessageDigest
//
//import org.apache.hadoop.conf.Configuration
//import org.apache.hadoop.fs.{FileSystem, Path}
//import org.apache.spark.sql.{SaveMode, SparkSession}
//
//object CheckMD5 {
//
//  def main(args: Array[String]): Unit = {
//
//    val tableName = args(0) //table name
//    val inFileHdfsDir = args(1) // input file (hdfs dir)
//    val count = args(2) // number of line for comparison
//    var whereExp = ""
//    if (args.length > 3) {
//      var dt = args(3) //bizDate
//      whereExp = " where dt = '" + dt + "'"
//    }
//
//    val outputDir = new Path(inFileHdfsDir + ".diff.log")
//
//    val conf = new Configuration
//    val hdfs: FileSystem = FileSystem.get(conf)
//    //删除已存在的差分日志目录
//    if (hdfs.exists(outputDir) && hdfs.isDirectory(outputDir)) {
//      hdfs.delete(outputDir, true)
//    }
//
//    val spark = SparkSession.builder()
//      //      .master("local")
//      .config("spark.sql.sources.partitionOverwriteMode", "dynamic")
//      .config("hive.exec.dynamic.partition.mode", "nonstrict")
//      //      .config("spark.sql.shuffle.partitions","20")
//      .appName("MD5ComparisonForFixedLen").enableHiveSupport().getOrCreate()
//
//    import spark.implicits._
//
//    val (colIdx, colData, colMd5) = ("idx", "data", "md5_coding")
//
//    // hive table data Df
//    val sqlFullData = spark.sql("SELECT * from " + tableName + whereExp).select("md5_coding")
//
//    // instantiation MD5 class
//    val md5 = new Md5Verifier()
//    //hdfs dir
//    val inFileRDD = spark.sparkContext.textFile(inFileHdfsDir)
//    val inFileMD5RDD = inFileRDD
//      .zipWithIndex().map(x => {
//      val data = x._1.split("@!@").map(s => md5.replaceDouble(s)).mkString
//      val md5Val = md5.md5Verify(data)
//      //      println("**************************************** md5Val=" + md5Val)
//      (x._2, x._1, md5Val) // idx, data, md5_coding
//    })
//
//    var limitCnt = inFileMD5RDD.count().toInt
//    if (count.toLong < limitCnt) {
//      limitCnt = count.toInt
//    }
//
//    val inFileMD5DF = inFileMD5RDD.toDF(colIdx, colData, colMd5)
//    val crossData = inFileMD5DF
//      .limit(limitCnt)
//      .join(sqlFullData, colMd5)
//      .map(x => {
//        //        val idxes = colNames.map(c=>x.fieldIndex(c))
//        (x.getLong(x.fieldIndex(colIdx)), x.getString(x.fieldIndex(colData)), x.getString(x.fieldIndex(colMd5)))
//      }).toDF(colIdx, colData, colMd5)
//    //    inFileMD5DF.show()
//    //    sqlFullData.show()
//    //    crossData.show()
//    val num = crossData.count()
//    if (num != limitCnt) {
//      val diffMd5Rdd = inFileMD5DF.join(
//        inFileMD5DF.limit(limitCnt).select(colIdx).except(crossData.select(colIdx)),
//        colIdx
//      ).rdd
//        .sortBy(x => x.getLong(x.fieldIndex(colIdx)))
//        .map(x => x.getString(x.fieldIndex(colData)))
//
//      //      val schema = StructType(colNames.map(fieldName => StructField(fieldName, StringType, true)))
//      //      val diffMd5Ds = spark.createDataFrame(diffMd5Rdd, schema)
//      val diffMd5Ds = diffMd5Rdd.toDF()
//      //      diffMd5Ds.show()
//      diffMd5Ds.write.mode(SaveMode.Overwrite).text(outputDir.toString)
//      //      System.exit(1)
//      //    } else {
//      //      System.exit(0)
//    }
//
//  }
//
//  class Md5Verifier(removeSpace: Boolean = true, delimiter: String = "") extends Serializable {
//    var _delimiter: String = delimiter
//    var _removeSpace: Boolean = removeSpace
//
//    def md5Verify(content: String): String = {
//      _md5Verify(trim(content))
//    }
//
//    def trim(content: String): String = {
//      var str = content
//      if (!_delimiter.isEmpty) {
//        str = removeStr(str, _delimiter)
//      }
//      if (_removeSpace) {
//        str = removeStr(str, " ")
//      }
//      str
//    }
//
//    private def _md5Verify(content: String): String = {
//      val md5 = MessageDigest.getInstance("MD5")
//      val dataMd5 = md5.digest((content).getBytes)
//      dataMd5.map("%02x".format(_)).mkString
//    }
//
//    def replaceDouble(content: String): String = {
//      var str = content
//      if (content.indexOf(".") > 0) {
//        str = content.replaceAll("0+?$", "").replaceAll("[.]$", "")
//      }
//      str
//    }
//
//    private def removeStr(content: String, removeStr: String): String = {
//      return content.replace(removeStr, "")
//    }
//  }
//
//
//}
